package com.game.common.concurrent;

import io.netty.util.concurrent.*;
import io.netty.util.internal.SystemPropertyUtil;

import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程池组
 */
public class GameEventExecutorGroup extends AbstractEventExecutorGroup {

    private final EventExecutor[] children;  //线程池
    private final AtomicInteger childIndex = new AtomicInteger();  //线程池索引
    private final AtomicInteger terminatedChildren = new AtomicInteger(); //线程池索引
    private final Promise<?> terminationFuture = new DefaultPromise<>(GlobalEventExecutor.INSTANCE); //线程池的关闭状态
    static final int DEFAULT_MAX_PENDING_EXECUTOR_TASKS = Math.max(16,
            SystemPropertyUtil.getInt("io.netty.eventexecutor.maxPendingTasks",
                    Integer.MAX_VALUE)); //线程池最大任务数

    public GameEventExecutorGroup(int nThreads) {
        this(nThreads, null);
    }
    public GameEventExecutorGroup(int nThreads, ThreadFactory threadFactory) {
        this(nThreads,
                threadFactory,
                DEFAULT_MAX_PENDING_EXECUTOR_TASKS,
                RejectedExecutionHandlers.reject());
    }

    /**
     * 构造线程池组
     * @param nThreads
     * @param threadFactory
     * @param maxPendingTasks
     * @param rejectedHandler
     */
    public GameEventExecutorGroup(int nThreads,
                                  ThreadFactory threadFactory,
                                  int maxPendingTasks,
                                  RejectedExecutionHandler rejectedHandler) {
        if (nThreads <= 0) {
            throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
        }
        if (threadFactory ==null) {
            threadFactory = newDefaultThreadFactory();
        }
        children = new SingleThreadEventExecutor[nThreads];
        for (int i = 0; i < nThreads; i++) {
            boolean success = false;
            try {
                children[i] = newChild(threadFactory, maxPendingTasks, rejectedHandler);
                success = true;
            } catch (Exception e) {
                // TODO: Think about if this is really what we want or if this should be a
                throw new IllegalStateException("failed to create a child event loop", e);
            } finally {
                if (!success) {
                    for (int j = 0; j < i; j++) {
                        children[j].shutdownGracefully();
                    }
                    for (int j = 0; j < i; j++) {
                        EventExecutor e = children[j];
                        try {
                            while (!e.isTerminated()) {
                                e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
                            }
                        } catch (InterruptedException ie) {
                            Thread.currentThread().interrupt();
                            break;
                        }
                    }
                }
            }
        }
        final FutureListener<Object> terminationListener = new FutureListener<>(){
            @Override
            public void operationComplete(Future<Object> objectFuture) throws Exception {
                if (terminatedChildren.incrementAndGet() == children.length) {
                    terminationFuture.setSuccess(null);
                }
            }
        };
        for (EventExecutor e: children) {
            e.terminationFuture().addListener(terminationListener);
        }
    }

    /**
     * 线程池工厂
     * @return
     */
    protected ThreadFactory newDefaultThreadFactory() {
        return new DefaultThreadFactory(getClass());
    }

    /**
     *
     * @param threadFactory
     * @param maxPendingTasks
     * @param rejectedHandler
     * @return
     * @throws Exception
     */
    protected EventExecutor newChild(ThreadFactory threadFactory,
                                     int maxPendingTasks,
                                     RejectedExecutionHandler rejectedHandler) throws Exception{
        return new DefaultEventExecutor(this, threadFactory, maxPendingTasks, rejectedHandler);
    }

    /**
     * 获取线程池
     * @param selectKey
     * @return
     */
    public EventExecutor select(Object selectKey) {
        if (selectKey == null) {
            throw new IllegalArgumentException("selectKey不能为空");
        }
        int hasCode = selectKey.hashCode();
        return getEventExecutor(hasCode);
    }

    /**
     * 提交任务
     * @param selectKey
     * @param task
     * @return
     * @param <T>
     */
    public <T> Future<T> submit(Object selectKey, Callable<T> task) {
        return this.select(selectKey).submit(task);
    }

    /**
     * 提交任务
     * @param selectKey
     * @param runnable
     * @return
     */
    public Future<?> submit(Object selectKey, Runnable runnable) {
        return this.select(selectKey).submit(runnable);
    }

    /**
     * 执行任务
     * @param selectKey
     * @param runnable
     */
    public void execute(Object selectKey, Runnable runnable) {
        this.select(selectKey).execute(runnable);
    }

    /**
     * 调度任务
     * @param selectKey
     * @param callable
     * @param delay
     * @param unit
     * @return
     * @param <V>
     */
    public <V> ScheduledFuture<V> schedule(Object selectKey, Callable<V> callable, long delay, TimeUnit unit) {
        return this.select(selectKey).schedule(callable, delay, unit);
    }

    /**
     * 调度任务
     * @param selectKey
     * @param runnable
     * @param delay
     * @param unit
     * @return
     */
    public ScheduledFuture<?> schedule(Object selectKey, Runnable runnable, long delay, TimeUnit unit) {
        return this.select(selectKey).schedule(runnable, delay, unit);
    }

    /**
     *
     * @param selectKey
     * @param command
     * @param initialDelay
     * @param period
     * @param unit
     * @return
     */
    public ScheduledFuture<?> scheduleAtFixedRate(Object selectKey,
                                                  Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit) {
        return this.select(selectKey).scheduleAtFixedRate(command, initialDelay, period, unit);
    }
    public ScheduledFuture<?> scheduleWithFixedDelay(Object selectKey,
                                                     Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit) {
        return this.select(selectKey).scheduleWithFixedDelay(command, initialDelay, delay, unit);
    }
    @Override
    public boolean isShuttingDown() {
        for (EventExecutor c : children) {
            if (!c.isShuttingDown()) {
                return false;
            }
        }
        return true;
    }

    @Override
    public Future<?> shutdownGracefully(long l, long l1, TimeUnit timeUnit) {
        for (EventExecutor c : children) {
            c.shutdownGracefully(l, l1, timeUnit);
        }
        return terminationFuture();
    }

    @Override
    public Future<?> terminationFuture() {
        return this.terminationFuture;
    }

    @Override
    public void shutdown() {
        this.shutdownGracefully();
    }

    @Override
    public boolean isShutdown() {
        for (EventExecutor c : children) {
            if (!c.isShutdown()) {
                return false;
            }
        }
        return true;
    }

    @Override
    public boolean isTerminated() {
        for (EventExecutor c : children) {
            if (!c.isTerminated()) {
                return false;
            }
        }
        return true;
    }

    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        long deadline = System.nanoTime() + unit.toNanos(timeout);
        loop: for (EventExecutor c : children) {
            for(;;) {
                long timeLeft = deadline - System.nanoTime();
                if (timeLeft <= 0) {
                    break loop;
                }
                if (c.awaitTermination(timeLeft, TimeUnit.NANOSECONDS)) {
                    break;
                }
            }
        }
        return isTerminated();
    }

    @Override
    public EventExecutor next() {
        return this.getEventExecutor(childIndex.getAndIncrement());
    }

    @Override
    public Iterator<EventExecutor> iterator() {
        return children().iterator();
    }

    protected Set<EventExecutor> children() {
        Set<EventExecutor> children = Collections.newSetFromMap(new LinkedHashMap<>());
        Collections.addAll(children, this.children);
        return children;
    }

    /**
     * 获得线程池
     * @param value
     * @return
     */
    private EventExecutor getEventExecutor(int value) {
        if (isPowerOfTwo(this.children.length)) {
            return children[value & children.length - 1];
        } else {
            return children[Math.abs(value % children.length)];
        }
    }

    /**
     * 判断是否是2的幂
     * @param value
     * @return
     */
    private boolean isPowerOfTwo(int value) {
        return (value & -value) == value;
    }

    public final int executorCount() {
        return this.children.length;
    }
}
